Add the flat nz map producer and its DELTA_BINARY consumer - #24501

…umer

Second of the nz map series. This one makes the feature do something, but
only under `LIBCUDF_PARQUET_NZ_MAP=1`; unset, nothing below runs and no
scratch is allocated.

The nz map is `nz_idx[rank] = input position of the rank-th valid value`, plus
its length `nz_count`. Naming it for the fields it holds keeps it clear of the
reader's own `pass` / `subpass`, which an earlier "prepass" spelling collided
with -- this file already has an unrelated string-size prepass.

Producer -- `precompute_nz_map` (decode_preprocess.cu) walks a claimed page's
already-decoded definition levels once, writing the leaf column's null mask and
the map, then records `nz_count`. A required page needs no map -- its map is the
identity -- and a null `nz_idx` is what tells the consumer to synthesize it.

The block-wide rank is taken from the ballot the loop already computes, rather
than from a `cub::BlockScan` over the same predicate, and the map pointer is
hoisted out of the loop: `PageInfo*` is not restrict-qualified, so each
iteration otherwise re-chased two dependent global loads to set up one 4-byte
store. Shared memory drops from 2708 to 2048 bytes per block and the kernel no
longer spills. At 40 registers and 128 threads it runs 12 blocks per SM, or 48
warps; the block scan was not what bounded that.

Consumer -- `decode_delta_binary_kernel_with_nz_map` reads that map and places
values without decoding levels, so it runs two warps rather than the
level-decoding kernel's three: a delta producer and a value writer. The writer trails the
producer by one batch, gated on `next_pass_start_idx()`. That lag is what lets
the two warps run concurrently in one phase instead of two phases split by a
barrier. The loop therefore exits only when neither side can advance: gating the
writer means a malformed page can leave `nz_count` permanently above anything
the delta stream can supply.

`decode_delta_binary_kernel` is untouched, including its name. Routing happens
through `non_nz_map_page_mask`, seeded on the host with the caller's page mask
and then cleared by the producer for every page it claims. The two kernels
therefore partition the pages because the producer's claim *is* the routing
decision, not because a second kernel re-derives it -- had it been re-derived,
the two could disagree and a page would be decoded twice or not at all. The
buffer is shared, so the further producers in this series clear their own claims
into it; a page has exactly one map kind, so the writes are disjoint.

For that partition to hold the consumer must not silently skip a page for any
*additional* reason, so the conditions the host already decides are documented
rather than re-tested. The one that cannot be decided on the host -- a claimed
page the producer wrote no count for -- is a `cudf_assert`, which catches it in
a debug build and compiles out in release. It is unreachable today: the producer
runs over a superset of the pages this kernel decodes.

`classify_nz_map_kind` names the encodings a consumer has been written for,
which is DELTA_BINARY_PACKED alone, and widens by one enumerator as each
remaining consumer lands. Claiming a page is what moves it off the
level-decoding kernel -- a claimed page carries a non-null `nz_map`, and that is
what routes it -- so the classifier names only the encodings that have somewhere
to go. A flat DELTA_BYTE_ARRAY page is equally decodable from a rank map and is
claimed once its consumer lands.

Stating that in the classifier rather than in a separate encoding mask keeps one
rule instead of two that have to agree.

The page mask is kept, rather than testing `has_nz_map()` inside
`decode_delta_binary_kernel` directly, so that kernel stays byte-identical while
the feature is opt-in -- turning the env var off is then a true rollback. Both
forms would read the same host-set fields, so this is not about the two
disagreeing. The in-code TODO records when to collapse it.

The producer gets its own stream and its own `cuda::event` rather than riding
`fork_streams`' shared one, so that only the kernel that reads its output waits
on it. The intent is that in a mixed file the PLAIN and dictionary decoders
overlap with it rather than queueing behind it; that overlap is not measured
here, since every benchmark below sets a single kernel mask.

Measured on `parquet_read_delta_binary_nullable_page_sizes` and
`parquet_read_delta_binary` (H100, clocks locked, feature off -> on):

  50% nulls          -25%
  required columns   -4.8%
  1% nulls           break-even

The win scales with how much level-decoding work the map removes. At 50% nulls
the bit-packed level walk it replaces is expensive; on required columns no map
is allocated at all and the gain is the narrower block alone. At 1% nulls the
levels are a cheap RLE run, so the map's own traffic -- 4 bytes per value
written, 4 per valid value read, against levels at 1-2 bytes per value -- costs
about what it saves. Narrowing the map's indices is the fix for that, and is
tracked for before the default flips.

Validation: PARQUET_TEST 563 passed / 1 pre-existing skip and HYBRID_SCAN_TEST
105/105, both with the nz map off and on. This is the first PR in the series
where the enabled arm decodes anything, so it is also the first real exercise of
the nz map under hybrid scan.

The test's env-var guard now restores any pre-existing value instead of unsetting
it. Unsetting meant that running the suite with `LIBCUDF_PARQUET_NZ_MAP=1`
-- which is how this feature is validated -- silently disabled it for every test
after this one, the second of 129 in the file. Measured over a full run: 758 of
1723 subpasses saw the feature enabled before the fix, 1721 after.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

"Claimed" was standing in for a plain fact the code already names: a page either
has an nz map or it does not, and `has_nz_map()` is the predicate. The word also
framed a selection rule as if pages were being contended for, when nothing
competes for them -- `classify_nz_map_kind` names the encodings a consumer has
been written for, and that is the whole decision.

Comments and doxygen only; no behaviour change. The unrelated uses elsewhere in
the reader are a different sense of the word and are left alone: a length that
"claims more bytes than remain in the page", `reader_impl_dict_transcode`
claiming ownership of a column, and the dictionary slot claims in chunk_dict.cu.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

`decode_delta_binary_kernel_with_nz_map` stops once neither of its two warps can
advance, which is what keeps a malformed page from spinning forever. But it then
returned without an error, and the producer has already written the validity bits
for the rows it had counted -- so a page whose delta stream holds fewer values
than its definition levels call for read back as valid rows over output slots
nothing ever wrote.

The stopping condition already knows this happened, so reporting it is one
comparison on thread 0 after the loop. It cannot fire on a well-formed page: the
only other way out of the loop is the `while` condition, which requires
`src_pos >= nz_count`.

This does not make a malformed page decode correctly, which is not something the
reader can do. It moves the failure from silently wrong data to a reported error.

`decode_delta_binary_kernel` has the same gap by a different route -- its writer
warp advances `src_pos` by a full batch whether or not the producer supplied
those values, so it runs to completion and writes stale rolling-buffer entries.
That is pre-existing and is left alone here, so the level-decoding path stays
byte-identical while the nz map is opt-in. Worth closing separately.

Raised by CodeRabbit on NVIDIA#24501.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

Four points from review on NVIDIA#24501, with the intrinsic modernization held back so
it can be measured on its own.

Drop `null_count_back_copier` from `precompute_flat_nz_map_kernel`. The producer
never writes a null count -- the only mention of one in the file was the copier's
own declaration -- so it copied back the value `setup_local_page_info`
initialized, which the consumer then overwrote.

`precompute_flat_nz_map` takes `cudf::memory_resources` and seeds the page mask
from `mr.get_temporary_mr()`. Behaviour is unchanged: the single-argument
constructor captures the current device resource for temporaries, which is what
the explicit `get_current_device_resource_ref()` was passing. What it buys is
that the temporary resource is now the caller's to choose.

`NzMapMatchesLevelDecoder` gains a required column. On that path the producer
writes no `nz_idx` and the consumer synthesizes the identity map from the rank
alone -- the test's comment claimed to cover it while the table held one nullable
column, so it never ran.

The same fixture now pins its page geometry through `max_page_fragment_size` and
`max_page_size_rows` rather than inheriting the writer's defaults, so what it
exercises does not move when those change.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

Two defects found reviewing the branch as a whole.

The truncated-stream report added earlier on this branch ORs its code onto
`s->setup.error` without checking whether that is already set. A page that fails
`setup_local_page_info` never enters the value loop, so `src_pos` stays 0 and the
test fires: the reader returns `EMPTY_PAGE | DATA_STREAM_OVERRUN` with the nz map
on and `EMPTY_PAGE` with it off, for the same file. Now reported only when
nothing else is.

`decode_delta_binary_kernel_with_nz_map` carried `skipped_leaf_values` handling
copied from the level-decoding kernel, where it is already dead --
`setup_local_page_info` zeroes it for any page without repetition, which is what
`classify_nz_map_kind` requires. It is also wrong for this kernel's data model:
`nz_count` counts from page position 0, so a rank already spans the prefix below
`first_row`, and `value_at(sp + skipped_leaf_values)` would have asked for that
prefix twice. The resume branch, the batch-size special case, the `target_pos`
subtraction and the index offset are gone, with the invariant stated where the
decode index is set up.

Also removed: a comment claiming the producer clears this page from a mask, which
described machinery deleted in the previous commit; a comment asserting the
level-decoding path "stays untouched", which stopped being true when it gained
its routing branch; a `__popc` the earlier modernization missed, one line below
its replaced sibling; an include left unused when the memory resource parameter
went away; and three `block.sync()` calls that each follow a thread-0 write to
global memory with nothing after them in the block.

`num_claimed` becomes `num_mapped`, which the prose sweep had already done
everywhere except this identifier.

Validation: PARQUET_TEST 563 passed / 1 pre-existing skip and HYBRID_SCAN_TEST
105/105, both with the nz map off and on.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

… reach

`NzMapMatchesLevelDecoder` read one nullable and one required flat column, which
left four paths unexercised.

An optional column with no null in it. `maybe_has_nulls` short-circuits on a
definition-level stream that is one repeated run at the max level, so the
producer takes its `!should_process_nulls` return with `nz_idx` allocated and the
null mask left to the all-valid fill. The previous validity pattern put a null in
every page, so this never ran.

A nested column, which `classify_nz_map_kind` declines. Its pages stay on the
level-decoding kernel while the flat columns go to the consumer, so the two
kernels now partition the pages of one file inside a test rather than only in an
nsys trace.

A chunked read. The producer runs once per output chunk over the same scratch,
each time with that chunk's row range rather than the pass's.

A slice landing mid-page. The bounded read skipped 1001 rows against 1000-row
pages, leaving `first_row` at 1, so the consumer's prefix search had a single
value below it; at 1500 it has 500.

The chunked comparison uses TABLES_EQUIVALENT because `cudf::concatenate` drops
an all-true validity mask, so the all-valid column comes back non-nullable while
the expected table is nullable. The values are identical.

Not covered: `level_type_size == 2`, which needs a max definition or repetition
level of 256 and so around that many levels of nesting. The `uint16_t`
instantiation of the producer is unreachable in practice for the same reason.

Validation: PARQUET_TEST 563 passed / 1 pre-existing skip and HYBRID_SCAN_TEST
105/105, both with the nz map off and on. An nsys trace of the test confirms the
producer, the consumer and the level-decoding kernel all run.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

…event

The producer and the one kernel that reads its output now share a stream, so
program order does what a `cuda::event` and an explicit wait did before. The
level-decoding kernel stops waiting with them: it reads `map_kind` and `nz_map`,
both host-set, and nothing the producer writes, so it never needed that edge.
Nested delta pages overlap the producer now rather than queueing behind it.

That makes the producer and the level-decoding kernel concurrent writers to the
leaf validity buffer. They own disjoint pages, and where their ranges meet inside
one bitmask word both are necessarily partial, which is the case `store_validity`
handles with atomics. racecheck over the delta and nz map tests with the feature
on reports no hazards.

No measurement is claimed here. The argument is that it removes a
synchronization object and lets one kernel overlap that previously could not; it
is not known to be faster, and the benchmarks that could show it are the ones
this series has repeatedly found too noisy to resolve changes of this size.

Also from review: `static_assert` that `level_decode_block_size` is a multiple of
the warp size and that the consumer's block is exactly the two warps its comment
claims; `cg::invoke_one` in place of `if (thread_rank() == 0)` where a block is
already in scope; the include left behind when the `cudf_assert` went; and the
chunked read test now sets both limits explicitly so it produces several
subpasses rather than relying on a default to do so.

Three comments trimmed. Review called them "AI thinking paste", which is fair:
they walked through how a design was reached rather than saying what the code
does. The aliasing reason for the writer-loop hoist and the rule about both
halves of the loop-exit test are kept, since neither is recoverable from the
code; the reasoning around them is not.

Validation: PARQUET_TEST 563 passed / 1 pre-existing skip and HYBRID_SCAN_TEST
106/106, both with the nz map off and on. compute-sanitizer racecheck and
memcheck: 0 hazards, 0 errors.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

Review of NVIDIA#24501 turned up four things.

`decode_delta_binary` launched the level-decoding kernel unconditionally, so
when every DELTA_BINARY page of a subpass carried a map, a full grid of blocks
started only to read the two fields that tell them to return. The host now also
computes whether any such page still needs level decoding, in the same walk that
decides whether to launch the consumer, so the two halves of the routing
decision cannot drift apart.

The nz map was sized from `num_input_values`, but the producer never indexes
past `min(num_input_values, num_decoded_level_values)`, and for a sliced read
the latter is smaller. A page whose levels fall entirely outside the row range
now gets no slice at all; the producer already reports a zero count for it and
the consumer never reaches the map.

`nz_map_stream` was taken from `streams.back()` even when no extra stream had
been forked, which aliased another decoder's stream, and would index an empty
vector if a subpass ever had no decode kernels.

The producer's docstring said it runs once per subpass. It runs once per output
chunk, each with its own row range; the allocation is what runs once per
subpass.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>